[SPARK-58736][SQL][PYTHON][CONNECT] Support incremental Python aggregators via Arrow - #57952

Add a Python analog of the Scala typed `Aggregator[IN, BUF, OUT]` with true
incremental (partial) aggregation. Users subclass `Aggregator`
(`zero`/`reduce`/`merge`/`finish` + `bufferSchema`) and wrap it with
`arrow_udaf(...)` for use in `groupBy().agg(...)`.

Unlike grouped-agg pandas/arrow UDFs (whole-group materialization), this is
planned as a two-stage aggregation with map-side combine: a PARTIAL stage folds
each group's input rows into a per-group Arrow buffer via `reduce`, the buffers
are shuffled by the grouping key, and a FINAL stage merges the partial buffers
via `merge` and produces the output via `finish`.

- New eval types SQL_GROUPED_AGG_ARROW_INCREMENTAL_PARTIAL/FINAL_UDF
  (Python `PythonEvalType` and JVM `PythonEvalType`).
- New Catalyst expression `PythonAggregate` carrying the intermediate buffer
  schema (unevaluable in the JVM, like `PythonUDAF`).
- New physical operators `PythonIncrementalAggregate{Partial,Final}Exec`,
  routed in `SparkStrategies` as Partial -> Exchange -> Final; the buffer
  crosses the shuffle as an Arrow struct column.
- Worker handlers: reduce-into-buffer (partial) and merge+finish (final).
- `arrow_udaf` / `Aggregator` API under `pyspark.sql.pandas.aggregator`.

Buffer schema is threaded to the JVM via a new nullable `bufferType` on
`UserDefinedPythonFunction`. Out of scope for now (follow-ups): distinct,
mixing with SQL aggregates, window/streaming, Spark Connect, SQL registration,
and a typed-vs-pickled buffer perf variant.

Co-authored-by: Isaac

… aggregator

Allow `spark.udf.register(name, udaf(agg))` so the incremental Python aggregator
can be invoked from SQL text (`SELECT my_agg(v) FROM t GROUP BY k`), matching
Scala's `spark.udf.register(name, functions.udaf(agg))`.

- Classic and Connect `UDFRegistration.register` accept
  SQL_GROUPED_AGG_ARROW_INCREMENTAL_FINAL_UDF and thread the buffer schema
  through (classic `register` reconstructs the UDF and would otherwise drop it;
  Connect passes it via `SparkConnectClient.register_udf` -> the PythonUDF proto).
- `udaf` sets `bufferSchema` on the returned wrapper too, so it survives
  registration. The Connect server already builds `PythonAggregate` in
  `handleRegisterUserDefinedFunction` via the shared `createUserDefinedPythonFunction`.
- Test: `test_sql_registration` in the shared mixin (runs classic + Connect).

Co-authored-by: Isaac

- Use PySparkNotImplementedError instead of a raw NotImplementedError in
  Aggregator.__call__ (PySpark custom-errors linter).
- Import have_pyarrow / pyarrow_requirement_message from pyspark.testing.utils
  (not sqlutils), which was causing the aggregator test modules to fail at import.

Co-authored-by: Isaac

…obal aggregation

Two blocking review items:

- PARTIAL/FINAL worker handlers now stream Arrow batches and fold them one at a
  time into the per-aggregator buffers, instead of `list(group)` + concatenating
  the whole group first. Map-side peak memory is bounded by a single batch (plus
  the buffers), not the whole group -- the point of the incremental API.
- A global (no-grouping) aggregation over empty input now returns the identity
  row `finish(zero)` instead of no row. GroupedPythonArrowInput cannot transmit
  an empty group, so the FINAL stage (which runs on a single AllTuples partition)
  injects one all-null buffer row; the worker skips null partial buffers and so
  merges nothing, yielding `finish(zero)`. Added a focused test
  (`df.limit(0).agg(udaf(...))`).

Co-authored-by: Isaac

…d names, doc/link fixes

- worker.py FINAL handler: type each output array explicitly via
  return_schema.field(i).type (mirroring PARTIAL) instead of relying on Arrow
  type inference + enforce_schema, so a non-trivial outputType (e.g. decimal)
  or an all-None column is robust.
- worker.py PARTIAL and FINAL: hoist the loop-invariant per-aggregator
  bufferSchema field-name lists (field_names_by_udf) out of the group/batch
  loops.
- PythonUDF.scala: fully-qualify the [[...PythonIncrementalAggregateExec]]
  Scaladoc link, which could not resolve from sql/catalyst (the class is in
  sql/core).
- aggregator.py: show null-input handling in the reduce example and document
  that udaf raises PySparkTypeError for a bad agg / bufferSchema.
- Tests: add a DecimalType-output aggregator test (exercises explicit output
  typing across the shuffle) and a null-input test; make the example Mean skip
  nulls to match SQL avg semantics.

Co-authored-by: Isaac